package transport.netty.client;

import codec.CommonDecoder;
import codec.CommonEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import serializer.Serializer;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/*
 * 初始化服务
 * */
public class ChannelProvider {
    private static final Logger logger = LoggerFactory.getLogger(ChannelProvider.class);
    private static Bootstrap bootstrap ;
    private static EventLoopGroup eventLoopGroup;
    //初始化 建立后连接
    static {
        bootstrap = new Bootstrap();
        eventLoopGroup = new NioEventLoopGroup();
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                //连接超时时间
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                //tcp 底层心跳机制
                .option(ChannelOption.SO_KEEPALIVE, true)
                //是否启用 nagle 算法，尽可能快的传送数据。
                .option(ChannelOption.TCP_NODELAY, true);
    }

    //使用一个 map集合 更加复用channel
    private static Map<String, Channel> channels = new ConcurrentHashMap<>();

    public static Channel getChannel(InetSocketAddress inetSocketAddress, Serializer serializer) {
        //从集合中取得 channel
        String key = inetSocketAddress.toString() + serializer.getCode();
        if (channels.containsKey(key)) {
            Channel channel = channels.get(key);
            if (channel != null && channel.isActive()) {
                return channel;
            } else {
                channels.remove(key);
            }
        }
        bootstrap.handler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel channel) throws Exception {
                channel.pipeline().addLast(new CommonEncoder(serializer))
//                        .addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS))
                        .addLast(new CommonDecoder())
                        .addLast(new NettyClientHandler());
            }
        });
        Channel channel = null;
        try {
            channel = connect(bootstrap, inetSocketAddress);
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        //加入集合 中以便复用
        channels.put(key,channel);
        return channel;
    }

    //根据地址返回一个channel 对象
    private static Channel connect(Bootstrap bootstrap, InetSocketAddress inetSocketAddress) throws ExecutionException, InterruptedException {
        CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
        // 异步 启动 一个客户端
        bootstrap.connect(inetSocketAddress).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                logger.info("客户端连接成功");
                completableFuture.complete(future.channel());
            }
        });
        return completableFuture.get();
    }
    public static void shutdownGracefully(){
        eventLoopGroup.shutdownGracefully();
    }

}
